#include "redis.hpp"
#include <iostream>

Redis::Redis()
    : _publish_context(nullptr)
    , _subscribe_context(nullptr)
{
}

Redis::~Redis()
{
    if(_publish_context)
    {
        redisFree(_publish_context);
    }
    if(_subscribe_context)
    {
        redisFree(_subscribe_context);
    }
}

// 连接 redis 服务器
bool Redis::connect()
{
    // 负责 publish 发布消息的上下文连接
    _publish_context = redisConnect("127.0.0.1", 6379);
    if(_publish_context == nullptr)
    {
        cerr << "connect redis failed!" << endl;
        return false;
    }

    // 负责 subscribe 订阅消息的上下文连接
    _subscribe_context = redisConnect("127.0.0.1", 6379);
    if(_subscribe_context == nullptr)
    {
        cerr << "connect redis failed!" << endl;
        return false;
    }

    // 在单独的线程中，监听通道上的事件，有消息给业务层上报
    thread t([&]() {
        observer_channel_message();
    });
    t.detach();

    cout << "connect redis-server success!" << endl;
    return true;
}

// 向 redis 指定的通道 channel 发布消息
bool Redis::publish(int channel, string messages)
{
    redisReply *reply = (redisReply *)redisCommand(_publish_context, "PUBLISH %d %s", channel, messages.c_str());
    if(reply == nullptr)
    {
        cerr << "publish command failed!" << endl;
        return false;
    }
    freeReplyObject(reply);  // 释放返回的结构体
    return true;
}

// 向 redis 指定的通道 subscribe 订阅消息
bool Redis::subscribe(int channel)
{
    /**
     * SUBSCRIBE命令本身会造成线程阻塞等待通道里面发生消息，这里只做订阅通道，不接收通道消息
     * 通道消息的接收专门在observer_channel message函数中的独立线程中进行
     * 只负责发送命令，不阻塞接收redis server响应消息，否则和notifyMsg线程抢占响应资源
    */

    if(redisAppendCommand(this->_subscribe_context, "SUBSCRIBE %d", channel) == REDIS_ERR)
    {
        cerr<<"subscribe command failed!" << endl;
        return false;
    }

    // redisBufferwrite可以循环发送缓冲区，直到缓冲区数据发送完毕 (done被置为1)
    int done =0;
    while(!done)
    {
        if(redisBufferWrite(this->_subscribe_context, &done) == REDIS_ERR)
        {
            cerr<<"subscribe command failed!"<< endl;
            return false;
        }
    }

    /**
     * 这里比 redisCommand 少了 redisGetReply，以阻塞方式等待远端响应
    */

    return true;
}

// 向 redis 指定的通道 unsubscribe 取消订阅消息
bool Redis::unsubscribe(int channel)
{
    if(redisAppendCommand(this->_subscribe_context, "UNSUBSCRIBE %d", channel) == REDIS_ERR)
    {
        cerr<<"unsubscribe command failed!" << endl;
        return false;
    }

    // redisBufferwrite可以循环发送缓冲区，直到缓冲区数据发送完毕 (done被置为1)
    int done =0;
    while(!done)
    {
        if(redisBufferWrite(this->_subscribe_context, &done) == REDIS_ERR)
        {
            cerr<<"unsubscribe command failed!"<< endl;
            return false;
        }
    }

    return true;
}

// 在独立的线程中接收订阅通道中的消息
void Redis::observer_channel_message()
{
    redisReply *reply = nullptr;
    while (redisGetReply(this->_subscribe_context, (void **)&reply) == REDIS_OK)  // 循环阻塞的方式等待消息
    {
        // 收到的消息是一个带三元素的数组，通道上发生事件一次性返回三个数据
        if(reply != nullptr && reply->element[2] != nullptr && reply->element[2]->str != nullptr)
        {
            // element[1] 通道号；element[2] 消息
            // 给业务层上报通道上发生的消息
            _notify_message_handler(atoi(reply->element[1]->str), reply->element[2]->str);
        }
        freeReplyObject(reply);
    }
    cerr << ">>>>>>>>>>>>>>> observer_channel_message_ quit <<<<<<<<<<<<<<<" << endl;
}

// 初始化向业务层上报通道消息的回调对象
void Redis::init_notify_handler(function<void(int, string)> fn)
{
    // 业务层注册回调
    this->_notify_message_handler = fn;
}